package com.asap.demo.sql;

import com.asap.demo.model.Bean;
import com.asap.demo.model.StandardEvent;
import com.asap.demo.utils.DateUtil;
import com.asap.demo.utils.Utils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;
import java.time.Instant;
import java.util.Date;
import java.util.Properties;

/**
 * @author wangbh
 * @Description: test
 * @date 2021/7/9 15:16
 */
public class FlinkSQLExample1 {
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.12","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:20:23.001Z","DEVICE_PARENT_TYPE":"WAF","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.13","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:21:24.001Z","DEVICE_PARENT_TYPE":"WAF","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.14","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:22:24.001Z","DEVICE_PARENT_TYPE":"WAF","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.16","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:23:24.001Z","DEVICE_PARENT_TYPE":"WAF","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.15","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:26:25.001Z","DEVICE_PARENT_TYPE":"WAF","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.11","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:15:21.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.12","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:20:23.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.13","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:21:24.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.14","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:22:24.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.16","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:23:24.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.15","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-21T18:26:25.001Z","DEVICE_PARENT_TYPE":"TDA","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.11","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:15:21.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.12","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:20:23.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.13","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:21:24.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.14","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:22:24.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.16","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:23:24.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}
//    {"ORG_ID":"1","EVENT_NAME":"爬虫","EVENT_THREE_TYPE":"20399","SRC_PORT":"123","DST_PORT":"124","DST_IP":"10.16.254.15","SRC_IP":"50.115.134.50","CREATE_TIME":"2021-07-20T18:26:25.001Z","DEVICE_PARENT_TYPE":"IPS","SNOW_ID":"85512","EVENT_THREE_TYPE_DESC":"端口扫描","ts":"2021-05-27T16:06:58Z","ACCOUNT":"asap","collectionName":"bwdOMS","eRuleId":"0","RULE_TJ_COUNT":11,"EVENT_ONE_TYPE":"20000","DIRECTION":"内部","EVENT_TWO_TYPE":"20100","EVENT_THREE_TYPE":"20101","DEVICE_TYPE":"OSM","DIRECTION":"0"}


   private static final Logger LOGGER = LoggerFactory.getLogger(FlinkSQLExample1.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment blinkStreamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
//		blinkStreamEnv.setParallelism(1);
        EnvironmentSettings blinkStreamSettings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment blinkStreamTableEnv = StreamTableEnvironment.create(blinkStreamEnv, blinkStreamSettings);
        Properties browseProperties = new Properties();
        browseProperties.put("bootstrap.servers", "192.168.1.25:9093");
        browseProperties.put("group.id", "temporal");
        browseProperties.put("auto.offset.reset", "latest");

        SingleOutputStreamOperator<StandardEvent> bStream = blinkStreamEnv
                .addSource(new FlinkKafkaConsumer<>(
                        "flink_pressure_test",
                        new SimpleStringSchema(),
                        browseProperties
                ))
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.minutes(1)) {
                    @Override
                    public long extractTimestamp(String element) {
                        return Timestamp.valueOf(Utils.transforDate(StandardEvent.parse(element).getField("CREATE_TIME"))).getTime();
                    }
                }).map(new MapFunction<String, StandardEvent>() {
                    @Override
                    public StandardEvent map(String value) throws Exception {
                        StandardEvent standardEvent=    StandardEvent.parse(value);
                        //standardEvent.getAllFields().replace("CREATE_TIME",Utils.transforDate(standardEvent.getField("CREATE_TIME")));
                        return standardEvent;
                    }
                });
        bStream.map(new MapFunction<StandardEvent, StandardEvent>() {
            @Override
            public StandardEvent map(StandardEvent value) throws Exception {
                LOGGER.debug(value.toString());
                return  value;
            }
        }).print();

        SingleOutputStreamOperator<Bean> bStream1=bStream.map(new MapFunction<StandardEvent,Bean>() {
            @Override
            public Bean map(StandardEvent standardEvent) throws Exception {
                Bean  bean=new Bean();
                bean.setDirectionDesc(standardEvent.getField("DIRECTION_DESC"));
                bean.setCreateTime(Instant.ofEpochMilli(DateUtil.parseDate(Utils.transforDate(standardEvent.getField("CREATE_TIME"))).getTime()));
                bean.setDeviceParentType(standardEvent.getField("DEVICE_PARENT_TYPE"));
                bean.setDstIp(standardEvent.getField("DST_IP"));
                bean.setEventTwoType(Long.valueOf(standardEvent.getField("EVENT_TWO_TYPE")));
                bean.setSnowId(Long.valueOf(standardEvent.getField("SNOW_ID")));
                bean.setSrcIp(standardEvent.getField("SRC_IP"));
                bean.setType(standardEvent.getField("TYPE"));
                bean.setSrcInfo(standardEvent.toString());
                bean.setEventName(standardEvent.getField("EVENT_NAME"));
             //   bean.setCreateTimeForTs(DateTime.parse((standardEvent.getField("CREATE_TIME"))));
                return bean;
            }
        });

     // Table inputTable = blinkStreamTableEnv.fromDataStream(bStream1,"type,eventName, directionDesc,srcIp,dstIp, createTime,deviceParentType,snowId,eventTwoType,ts,srcInfo,createTimeForTs");
        Table inputTable = blinkStreamTableEnv.fromDataStream(bStream1,
                Schema.newBuilder()
                        .column("eventTwoType", "BIGINT")
                        .column("deviceParentType", "STRING")
                        .column("type", "STRING")
                        .column("eventName", "STRING")
                        .column("directionDesc", "STRING")
                        .column("srcIp", "STRING")
                        .column("dstIp", "STRING")
                        .column("createTime", "TIMESTAMP_LTZ(3)")
                        .column("snowId", "BIGINT")
                        .column("srcInfo", "STRING")
                        .watermark("createTime", "SOURCE_WATERMARK()")
                        .build());
        blinkStreamTableEnv.createTemporaryView("InputTable", inputTable);
        Table resultTable = blinkStreamTableEnv.sqlQuery("SELECT * FROM InputTable");
       // blinkStreamTableEnv.toRetractStream(resultTable, Row.class).print("query==");

        String querySQL = "select deviceParentType,srcIp,dstIp,srcInfo from InputTable where deviceParentType='WAF' AND eventName like '%爬虫%' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceParentType,srcIp,dstIp,srcInfo";
        Table table1 = blinkStreamTableEnv.sqlQuery(querySQL);
       // blinkStreamTableEnv.toRetractStream(table1, Row.class).print("query1==");

        String querySQL2 = "select deviceParentType,srcIp,dstIp,srcInfo from InputTable where deviceParentType='TDA' AND eventName like '%爬虫%' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceParentType,srcIp,dstIp,srcInfo";
        Table table2 = blinkStreamTableEnv.sqlQuery(querySQL2);
       // blinkStreamTableEnv.toRetractStream(table2, Row.class).print("query2==");

        String querySQL3 = "select deviceParentType,srcIp,dstIp,srcInfo from InputTable where deviceParentType='IPS' AND eventName like '%爬虫%' group by TUMBLE(createTime, INTERVAL '5' MINUTE),deviceParentType,srcIp,dstIp,srcInfo";
        Table table3 = blinkStreamTableEnv.sqlQuery(querySQL3);
       // blinkStreamTableEnv.toRetractStream(table3, Row.class).print("query3==");

        blinkStreamTableEnv.createTemporaryView("table1", table1);
        blinkStreamTableEnv.createTemporaryView("table2", table2);
        blinkStreamTableEnv.createTemporaryView("table3", table3);
        String querySQL8 = "select * from table2 ,table1,table3 where table1.srcIp=table2.srcIp and table1.dstIp=table3.dstIp";
        Table table8 = blinkStreamTableEnv.sqlQuery(querySQL8);



        blinkStreamTableEnv.toRetractStream(table8, Row.class).print("query8==");

        //DataStream<Row> dsRow = blinkStreamTableEnv.toAppendStream(table8, Row.class);
       // dsRow.addSink()

        blinkStreamEnv.execute("Blink Stream SQL distinct demo");
    }
}
